import ast
from datetime import datetime, timedelta
import threading
from impala.dbapi import connect
import time
class Presto(object):
    __instance = None
    def __new__(cls, *args, **kwargs):
        if cls.__instance is None:
            cls.__instance = super(Presto, cls).__new__(cls, *args, **kwargs)
        return cls.__instance
    def result_iterator(self, cursor, buffer=1000):
        while True:
            results = cursor.fetchmany(buffer)
            if not results:
                break
            for result in results:
                yield result
    def run_sync(self, sql, is_fetch, engine_type, is_farlight=False):
        """
        同步请求,能满足大多数使用场景
        :param sql: 需要执行的SQL
        :param is_fetch: 是否需要拉取结果
        :param engine_type: 执行sql引擎类型
        :return:
        """
        config = {
            'host': 'bbx.lilithgame.com',  # hiveserver2的连接地址
            'port': 10000,  # hiveserver2的连接端口
            'user': "0734999153790a93eae2854f6ea4909dShyuBgTVBsv",  # 用户名  api_key/user
            'password': 'jTesP5YwFnKto8ZN250jyB7FdUW8WA7uHlDqCwdIU2',  # 密码  api_secret/id_token
            "auth_mechanism": 'plain',  # 认证方式，此处固定为示例值
            "engine_type": engine_type,  # 查询引擎类型 trino/impala/hive
            "query_operator": "guohuiliang",  # sql执行操作者，推荐使用公司邮箱前缀
            "timeout": 600  # sql执行单次RPC调用超时时间
        }
        if is_farlight:
            config['host'] = 'bbx.farlightgames.com'
        conn = connect(**config)
        cursor = conn.cursor()
        # 建立连接
        # 异步执行sql
        # print(f"Execute sql:{sql}")

        try:
            cursor.execute(sql)
            # 拉取结果
            if is_fetch:
                res_data_list = []
                # print(cursor.description)
                for row in self.result_iterator(cursor):
                    # print(row)
                    res_data_list.append(row)
                return res_data_list
        except Exception as e:
            print(e)
        finally:
            cursor.close()
            conn.close()

    def run_async(self, sql, is_fetch, engine_type, is_farlight=False):
        """
        异步请求,可获取执行中间的状态信息
        :param sql: 需要执行的SQL
        :param is_fetch: 是否有结果
        :param engine_type: 执行sql引擎类型
        :return:
        """
        # 连接配置
        config = {
            'host': 'bbx.lilithgame.com',  # hiveserver2的连接地址
            'port': 10000,  # hiveserver2的连接端口
            'user': "0734999153790a93eae2854f6ea4909dShyuBgTVBsv",  # 用户名  api_key/user
            'password': 'jTesP5YwFnKto8ZN250jyB7FdUW8WA7uHlDqCwdIU2',  # 密码  api_secret/id_token
            "auth_mechanism": 'plain',  # 认证方式，此处固定为示例值
            "engine_type": engine_type,  # 查询引擎类型 trino/impala/hive
            "query_operator": "guohuiliang",  # sql执行操作者，推荐使用公司邮箱前缀
            "timeout": 600  # sql执行单次RPC调用超时时间
        }
        if is_farlight:
            config['host'] = 'bbx.farlightgames.com'
        # 建立连接
        conn = connect(**config)
        cursor = conn.cursor()
        # 异步执行sql
        # print(f"Execute sql:{sql}")
        try:
            cursor.execute_async(sql)
            # 轮询状态
            loop_start = time.time()
            while True:
                (execute_state, execute_process) = cursor.get_execute_state()
                print(f"execute_state: {execute_state},execute_process: {execute_process}")
                if not cursor.op_state_is_executing(execute_state):
                    break
                time.sleep(cursor.get_sleep_interval(loop_start))
            # 拉取结果
            if is_fetch:
                # print(cursor.description)
                res_data_list = []
                for row in self.result_iterator(cursor):
                    # print(row)
                    res_data_list.append(row)
                return res_data_list
        except Exception as e:
            print(e)
        finally:
            cursor.close()
            conn.close()
    def get(self, sql, is_fetch=True, engine_type='trino'):
        res_list = self.run_sync(sql, is_fetch, engine_type)
        return res_list


config = {
    'host': 'bbx.lilithgame.com',  # hiveserver2的连接地址
    'port': 10000,  # hiveserver2的连接端口
    'user': "0734999153790a93eae2854f6ea4909dShyuBgTVBsv",  # 用户名  api_key/user
    'password': 'jTesP5YwFnKto8ZN250jyB7FdUW8WA7uHlDqCwdIU2',  # 密码  api_secret/id_token
    "auth_mechanism": 'plain',  # 认证方式，此处固定为示例值
    "engine_type": 'trino',  # 查询引擎类型 trino/impala/hive
    "query_operator": "guohuiliang",  # sql执行操作者，推荐使用公司邮箱前缀
    "timeout": 600  # sql执行单次RPC调用超时时间
}



def result_iterator( cursor, buffer=1000):
    while True:
        results = cursor.fetchmany(buffer)
        if not results:
            break
        for result in results:
            yield result

INSERT_LOCK = threading.Lock()
RUN_LOCK = threading.Lock()
def run_sql(sql):
    retries = 0
    with RUN_LOCK:
        while retries < 3:
            try:
                # 建立连接
                conn = connect(**config)
                cursor = conn.cursor()
                cursor.execute_async(sql)
                # 轮询状态
                loop_start = time.time()
                while True:
                    (execute_state, execute_process) = cursor.get_execute_state()
                    print(f"execute_state: {execute_state},execute_process: {execute_process}")
                    if not cursor.op_state_is_executing(execute_state):
                        break
                    time.sleep(cursor.get_sleep_interval(loop_start))
                # 拉取结果
                if True:
                    res_data_list = []
                    rows = result_iterator(cursor)
                    if not rows:
                        return []
                    for row in rows:
                        res_data_list.append(row)
                    return res_data_list
            except Exception as e:
                if type(e).__name__ == "ProgrammingError":
                    print(f"Error Type:{type(e).__name__}")
                    print(f"Error:{e}")
                    print(f"无拉取结果 ......")
                    return 
                
                print(f"Error Type:{type(e).__name__}")
                print(f"Error:{e}")
                print("上传数据库出错 ......")
                retries += 1


def run_sql_sync(sql):
    retries = 0
    with INSERT_LOCK:
        while retries < 3:
            try:
                # 建立连接
                conn = connect(**config)
                cursor = conn.cursor()
                cursor.execute(sql)
                # 拉取结果
                if True:
                    res_data_list = []
                    for row in result_iterator(cursor):
                        res_data_list.append(row)
                    return res_data_list
            except Exception as e:
                if type(e).__name__ == "ProgrammingError":
                    print(f"Error Type:{type(e).__name__}")
                    print(f"Error:{e}")
                    print(f"无拉取结果 ......")
                    return 
                
                print(f"Error Type:{type(e).__name__}")
                print(f"Error:{e}")
                print("上传数据库出错 ......")
                retries += 1

def insert_bilibili_user_detail(up_detail):
    presto_t = Presto()
    gender = ""
    if up_detail["sex"] == "女":
        gender = "female"
    elif up_detail["sex"] == "男":
        gender = "male"
    else:
        gender = "unknown"

    sql = f"""
        INSERT INTO iceberg.dap_dw.koc_creators
        VALUES
        ('{str(datetime.now().date())}',
        'bilibili',
        '{up_detail["uid"]}',
        '{up_detail["nick_name"].replace("'", "''")}',
        '{up_detail["link"].replace("'", "''")}',
        '{up_detail["avatar"].replace("'", "''")}',
        '',
        '',
        '{up_detail["fans"]}',
        '',
        '',
        '{gender}',
        '{up_detail["register_date"]}',
        DATE_TRUNC('second',CURRENT_TIMESTAMP),
        DATE_TRUNC('second',CURRENT_TIMESTAMP),
        '{up_detail["tags"].replace("'", "''")}',
        '{up_detail["seasons_series"].replace("'","''")}',
        '')
    """

    run_sql_sync(sql)

def insert_bilibili_video_details(video_details, up_detail):
    gender = ""
    if up_detail["sex"] == "女":
        gender = "female"
    elif up_detail["sex"] == "男":
        gender = "male"
    else:
        gender = "unknown"

    value_list = []
    batch_size = 600  # 每批插入600条
    total_videos = len(video_details)
    p = 1
    try:
        for i, video_detail in enumerate(video_details, 1):
            value = f"""
                (
                '{str(datetime.now().date())}',
                'bilibili',
                '{video_detail["uid"]}',
                '{video_detail["bvid"]}',
                '{video_detail["title"].replace("'", "''")}',
                '{video_detail["video_link"].replace("'", "''")}',
                '{video_detail["view"]}',
                '{video_detail["like"]}',
                '{video_detail["share"]}',
                '{video_detail["reply"]}',
                '0',
                '0',
                '{video_detail["favorite"]}',
                '{video_detail["coin"]}',
                '{video_detail["danmaku"]}',
                '',
                '',
                '{up_detail["fans"]}',
                '',
                '',
                '{gender}',
                '{video_detail["pubdate"]}',
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                '{video_detail["tags"].replace("'", "''")}',
                '{video_detail["duration"]}',
                ''
                )
            """
            value_list.append(value)

            if i % batch_size == 0 or i == total_videos:
                sql = f"""
                    INSERT INTO iceberg.dap_dw.koc_posts
                    VALUES
                    {",".join(value_list)}
                """
  
                run_sql_sync(sql)
                value_list = []  # 清空当前批次
                print(f"第{p}批已上传完成......")
                p += 1
    except Exception as e:
        print(e)
        return




def insert_douyin_user_detail(user_detail):
    try:
        gender = "unknown"
        sql = f"""
            INSERT INTO iceberg.dap_dw.koc_creators
            VALUES
            ('{str(datetime.now().date())}',
            'douyin',
            '{user_detail["sec_uid"]}',
            '{user_detail["nickname"].replace("'", "''")}',
            '{user_detail["home_page_url"].replace("'", "''")}',
            '{user_detail["avatar_url"].replace("'", "''")}',
            '',
            '{user_detail["ip_location"]}',
            '{user_detail["follower_count"]}',
            '',
            '',
            '{gender}',
            '{user_detail["register_time"]}',
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            '{user_detail["tags"].replace("'", "''")}',
            '{user_detail["seasons_series"].replace("'","''")}',
            '{user_detail["origin_id"]}'
            )
        """
        # presto_t.run_sync(sql, False, 'trino')
        run_sql_sync(sql)
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return


def insert_douyin_video_details(video_details, user_detail):
    gender = "unknown"
    value_list = []
    batch_size = 600
    total_videos = len(video_details)
    try:
        for i, video_detail in enumerate(video_details, 1):
            value = f"""
                (
                '{str(datetime.now().date())}',
                'douyin',
                '{video_detail["sec_uid"]}',
                '{video_detail["aweme_id"]}',
                '{video_detail["title"].replace("'", "''")}',
                '{video_detail["video_link"].replace("'", "''")}',
                '0',
                '{video_detail["like"]}',
                '{video_detail["share"]}',
                '{video_detail["comment"]}',
                '0',
                '0',
                '{video_detail["collect"]}',
                '0',
                '0',
                '',
                '{user_detail["ip_location"]}',
                '{user_detail["follower_count"]}',
                '',
                '',
                '{gender}',
                '{video_detail["create_time"]}',
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                '{video_detail["tags"].replace("'", "''")}',
                '{video_detail["duration"]}',
                '{user_detail["origin_id"]}'
                )
            """
            value_list.append(value)


            if i % batch_size == 0 or i == total_videos:
                sql = f"""
                    INSERT INTO iceberg.dap_dw.koc_posts
                    VALUES
                    {",".join(value_list)}
                """
                # presto_t.run_sync(sql, False, 'trino')
                run_sql_sync(sql)
                value_list = []  # 清空当前批次
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return


def insert_kuaishou_user_detail(user_detail):
    try:
        gender = "unknown"
        if user_detail["gender"] == 'M':
            gender = "男"
        elif user_detail["gender"] == 'F':
            gender = "女"
        sql = f"""
            INSERT INTO iceberg.dap_dw.koc_creators
            VALUES
            ('{str(datetime.now().date())}',
            'kuaishou',
            '{user_detail["user_id"]}',
            '{user_detail["user_name"].replace("'", "''")}',
            '{user_detail["home_url"].replace("'", "''")}',
            '{user_detail["avatar_url"].replace("'", "''")}',
            '',
            '',
            '{user_detail["fans"]}',
            '',
            '',
            '{gender}',
            '{user_detail["register_time"]}',
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            '{user_detail["tags"].replace("'", "''")}',
            '',
            ''
            )
        """
        # presto_t.run_sync(sql, False, 'trino')
        run_sql_sync(sql)
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return

def insert_kuaishou_video_details(video_details, user_detail):
    value_list = []
    batch_size = 600
    total_videos = len(video_details)
    try:
        gender = "unknown"
        if user_detail["gender"] == 'M':
            gender = "男"
        elif user_detail["gender"] == 'F':
            gender = "女"
        for i, video_detail in enumerate(video_details, 1):
            value = f"""
                (
                '{str(datetime.now().date())}',
                'kuaishou',
                '{video_detail["kuaishou_user_id"]}',
                '{video_detail["video_id"]}',
                '{video_detail["title"].replace("'", "''")}',
                '{video_detail["video_link"].replace("'", "''")}',
                '{video_detail["play_count"]}',
                '{video_detail["like_count"]}',
                '0',
                '0',
                '0',
                '0',
                '0',
                '0',
                '0',
                '',
                '',
                '{user_detail["fans"]}',
                '',
                '',
                '{gender}',
                '{video_detail["create_time"]}',
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                '{video_detail["tags"].replace("'", "''")}',
                '{video_detail["duration"]}',
                ''
                )
            """
            value_list.append(value)


            if i % batch_size == 0 or i == total_videos:
                sql = f"""
                    INSERT INTO iceberg.dap_dw.koc_posts
                    VALUES
                    {",".join(value_list)}
                """
                # presto_t.run_sync(sql, False, 'trino')
                run_sql_sync(sql)
                value_list = []  # 清空当前批次
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return

def insert_xiaohongshu_user_detail(user_detail):
    try:
        gender = "unknown"
        if user_detail["gender"] == 'female':
            gender = "男"
        elif user_detail["gender"] == 'male':
            gender = "女"
        sql = f"""
            INSERT INTO iceberg.dap_dw.koc_creators
            VALUES
            ('{str(datetime.now().date())}',
            'xiaohongshu',
            '{user_detail["user_id"]}',
            '{user_detail["nickname"].replace("'", "''")}',
            '{user_detail["home_url"].replace("'", "''")}',
            '{user_detail["avatar"].replace("'", "''")}',
            '',
            '{user_detail["ip_location"]}',
            '{user_detail["fans"]}',
            '',
            '',
            '{gender}',
            '{user_detail["register_time"]}',
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            DATE_TRUNC('second',CURRENT_TIMESTAMP),
            '{user_detail["tags"].replace("'", "''")}',
            '',
            '{user_detail["origin_id"]}'
            )
        """
        # presto_t.run_sync(sql, False, 'trino')
        run_sql_sync(sql)
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return

def insert_xiaohongshu_video_details(video_details, user_detail):
    value_list = []
    batch_size = 600
    total_videos = len(video_details)
    try:
        gender = "unknown"
        if user_detail["gender"] == 'female':
            gender = "男"
        elif user_detail["gender"] == 'male':
            gender = "女"
        for i, video_detail in enumerate(video_details, 1):
            value = f"""
                (
                '{str(datetime.now().date())}',
                'xiaohongshu',
                '{user_detail["user_id"]}',
                '{video_detail["note_id"]}',
                '{video_detail["title"].replace("'", "''")}',
                '{video_detail["note_url"].replace("'", "''")}',
                '0',
                '{video_detail["like_count"]}',
                '{video_detail["share_count"]}',
                '{video_detail["comment_count"]}',
                '0',
                '0',
                '{video_detail["collect_count"]}',
                '0',
                '0',
                '',
                '{user_detail["ip_location"]}',
                '{user_detail["fans"]}',
                '',
                '',
                '{gender}',
                '{video_detail["create_time"]}',
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                DATE_TRUNC('second',CURRENT_TIMESTAMP),
                '{video_detail["tags"].replace("'", "''")}',
                '{video_detail["duration"]}',
                '{user_detail["origin_id"]}'
                )
            """
            value_list.append(value)


            if i % batch_size == 0 or i == total_videos:
                sql = f"""
                    INSERT INTO iceberg.dap_dw.koc_posts
                    VALUES
                    {",".join(value_list)}
                """
                # presto_t.run_sync(sql, False, 'trino')
                run_sql_sync(sql)
                value_list = []  # 清空当前批次
    except Exception as e:
        print(f"上传数据库出错:{e}")
        return


def get_creators_by_dt_bilibili():
    creator_id_list = []
    sql = f"""
        select creator_id from iceberg.dap_dw.koc_creators
        where dt = '{get_dt()}' and media = 'bilibili'
    """
    rows = run_sql(sql)
    for row in rows:
        creator_id_list.append(row[0])
    return creator_id_list

def get_creators_by_dt_media(media):
    creator_id_list = []
    sql = f"""
        select creator_id from iceberg.dap_dw.koc_creators
        where dt = '{get_dt()}' and media = '{media}'
    """
    rows = run_sql(sql)
    if rows is None:
        return creator_id_list
    for row in rows:
        creator_id_list.append(row[0])
    return creator_id_list

def is_added_by_mid(mid):
    presto_t = Presto()
    sql = f"""
        select count(*) from iceberg.dap_dw.koc_creators
        where creator_id = '{mid}' and dt = '{get_dt()}'
    """
    rows = run_sql(sql)
    for row in rows:
        if row[0] > 0:
            return True
        else:
            return False

def get_dt():
    return str(datetime.now().date())


def get_previous_dt(media):
    try:
        sql = f"""
            SELECT *
            FROM iceberg.dap_dw.koc_logs
            WHERE media = '{media}'
            ORDER BY CAST(dt AS DATE) DESC
        """
        rows = run_sql(sql)
        return rows[0][0]
    except Exception as e:
        print(e)
        return ""

def get_previous_day_update_creators_by_media(media):
    previous_dt = get_previous_dt(media)
    if previous_dt == "":
        return []
    previous_day_creators = []

    sql = f"""
        select creator_id from iceberg.dap_dw.koc_creators
        where dt = '{previous_dt}' and media = '{media}'
    """
    rows = run_sql(sql)
    for row in rows:
        previous_day_creators.append(row[0])
    return previous_day_creators

def get_this_day_update_creators_by_media(media):
    dt = get_dt()
    this_day_creators = []
    sql = f"""
        select creator_id from iceberg.dap_dw.koc_creators
        where dt = '{dt}' and media = '{media}'
    """
    rows = run_sql(sql)
    for row in rows:
        this_day_creators.append(row[0])
    return this_day_creators
def get_tags_by_id_and_media(id,media):
    sql = f"""
        select dt,tags from iceberg.dap_dw.koc_creators
        where creator_id = '{id}' and media = '{media}'
        order by cast(dt as date) desc
    """
    rows = run_sql(sql)
    for row in rows:
        if row[1] is None:
            return []
        return str_list_to_list(row[1])

def str_list_to_list(str_list) -> list:
    try:
        real_list = ast.literal_eval(str_list)
    except Exception as e:
        print(e)
        real_list = []
    return real_list


def update_tags_by_id_and_media(tags,id,media):
    tags_str = str(tags)
    sql = f"""
        update iceberg.dap_dw.koc_creators
        set tags = '{tags_str.replace("'", "''")}'
        where creator_id = '{id}' and media = '{media}'
    """
    run_sql_sync(sql)


def insert_success_log_by_media(media):
    dt = get_dt()
    sql = f"""
        insert into iceberg.dap_dw.koc_logs
        values ('{dt}','{media}','SUCCESS')
    """
    run_sql_sync(sql)


def main():
    print(get_tags_by_id_and_media('395153972','bilibili'))


    # tags = ['英雄联盟']
    # update_tags_by_id_and_media(tags,'11423506','bilibili')

    # sql = "select * from iceberg.dap_dw.koc_posts where creator_id = '11073'"
    dt = get_dt()
    # get_previous_day_update_creators('bilibili')
    # print(get_previous_day_update_creators())
    # sql = f"select count(*) from iceberg.dap_dw.koc_creators where dt = '2025-05-12' and media = 'bilibili'"
    # rows = run_sql(sql)
    # for row in rows:
    #     print(row)
    # sql = "select creator_id from iceberg.dap_dw.koc_creators"
    #
    # all_creator_id_list = []
    # rows = run_sql(sql)
    # for row in rows:
    #     all_creator_id_list.append(row[0])
    #     print(row)
    # print(len(all_creator_id_list))
    #


if __name__ == '__main__':
    list = str_list_to_list("[''英雄联盟'']")
    print(list[0])


